package com.cennavi.funcs;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;

import com.cennavi.beans.DevAndLane;
import com.cennavi.beans.DevAndRtic;
import com.cennavi.beans.FlowAndTime_Rtic;
import com.cennavi.beans.SensorFlowData;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月18日 下午2:39:49 
    * @version 1.0 
    */

public class Data2BeanPairFlatMapFunction implements PairFlatMapFunction<Tuple2<String, String>, String, FlowAndTime_Rtic> {

	private static final long serialVersionUID = 1L;
	private HashMap<String, List<DevAndLane>> dev_laneMap;
	private HashMap<String,DevAndRtic> dev_rticMap;
	
	//构造函数
	public Data2BeanPairFlatMapFunction(
			Broadcast<HashMap<String, List<DevAndLane>>> dev_laneBroadMap,
			Broadcast<HashMap<String, DevAndRtic>> dev_rticBroadMap){
		this.dev_laneMap = dev_laneBroadMap.value();
		this.dev_rticMap = dev_rticBroadMap.value();
	}
	
	public Iterable<Tuple2<String, FlowAndTime_Rtic>> call(Tuple2<String, String> arg0) throws Exception {
		// 
		ArrayList<Tuple2<String, FlowAndTime_Rtic>> returnList = new ArrayList<Tuple2<String, FlowAndTime_Rtic>>();
		//取出每一行的value，按照逗号打断
		String[] data = arg0._2.split(",");
		SensorFlowData sensorFlowData = new SensorFlowData();
		sensorFlowData.setDev_id(data[0]);
		sensorFlowData.setDev_way_id(data[1]);
		sensorFlowData.setFlow(Integer.parseInt(data[2]));
		sensorFlowData.setCollect_time(data[3]);
		
		//把每条数据匹配到rtic上面，rticid作为key
		String rtic_id = "";
		FlowAndTime_Rtic flowAndTime_Rtic = new FlowAndTime_Rtic();
		flowAndTime_Rtic.setFlow(sensorFlowData.getFlow());
		flowAndTime_Rtic.setCollect_time(sensorFlowData.getCollect_time());
		
		if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic_num().equals("1")){
			rtic_id = this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic1_id();
			returnList.add(new Tuple2<String, FlowAndTime_Rtic>(rtic_id,flowAndTime_Rtic));
		}
		
		returnList.add(new Tuple2<String, FlowAndTime_Rtic>(rtic_id, flowAndTime_Rtic));
		return returnList;
	}

}
